package com.cyzy.sse;

import okhttp3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * EventSource的实际实现类，用于处理SSE连接
 */
public class RealEventSource implements EventSource {
    private static final Logger logger = LoggerFactory.getLogger(RealEventSource.class);

    private final Request originalRequest;
    private final EventSourceListener listener;
    private Call call;
    private State state = State.CONNECTING;

    public RealEventSource(Request request, EventSourceListener listener) {
        if (!"GET".equals(request.method())) {
            throw new IllegalArgumentException("请求方法必须是GET");
        }
        this.originalRequest = request;
        this.listener = listener;
    }

    @Override
    public void connect(OkHttpClient client) {
        if (state != State.CONNECTING) {
            throw new IllegalStateException("已经连接或已关闭");
        }

        // 添加必要的请求头
        Request request = originalRequest.newBuilder()
                .header("Accept", "text/event-stream")
                .header("Cache-Control", "no-cache")
                .build();

        call = client.newCall(request);
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                notifyFailure(e, null);
            }

            @Override
            public void onResponse(Call call, Response response) {
                if (!response.isSuccessful()) {
                    notifyFailure(new IOException("非成功HTTP响应: " + response.code()), response);
                    return;
                }

                ResponseBody body = response.body();
                if (body == null) {
                    notifyFailure(new IOException("响应体为空"), response);
                    return;
                }

                try {
                    state = State.OPEN;
                    listener.onOpen(RealEventSource.this, response);
                    processEvents(response);
                } catch (Exception e) {
                    notifyFailure(e, response);
                }
            }
        });
    }

    private void processEvents(Response response) throws IOException {
        ResponseBody body = response.body();
        if (body == null) {
            return;
        }

        String line;
        String id = null;
        String type = null;
        StringBuilder data = new StringBuilder();

        try (okio.BufferedSource source = body.source()) {
            while (!source.exhausted() && state != State.CLOSED && state != State.SHUTDOWN) {
                line = source.readUtf8LineStrict();
                
                if (line.isEmpty()) {
                    // 空行表示事件边界，发送累积的事件
                    if (data.length() > 0) {
                        String eventData = data.toString();
                        listener.onEvent(this, id, type, eventData);
                        id = null;
                        type = null;
                        data.setLength(0);
                    }
                } else if (line.startsWith("data:")) {
                    // 数据行
                    String value = line.substring(5).trim();
                    data.append(value);
                } else if (line.startsWith("id:")) {
                    // 事件ID
                    id = line.substring(3).trim();
                } else if (line.startsWith("event:")) {
                    // 事件类型
                    type = line.substring(6).trim();
                } else if (line.startsWith(":")) {
                    // 注释行，忽略
                } else {
                    // 其他格式的行，忽略
                    logger.debug("忽略未知格式的SSE行: {}", line);
                }
            }
        } catch (IOException e) {
            if (state != State.CLOSED && state != State.SHUTDOWN) {
                throw e;
            }
        } finally {
            close();
        }
    }

    private void notifyFailure(Throwable t, Response response) {
        if (state == State.CLOSED || state == State.SHUTDOWN) {
            return;
        }
        state = State.CLOSED;
        listener.onFailure(this, t, response);
    }

    @Override
    public void cancel() {
        if (state == State.CLOSED || state == State.SHUTDOWN) {
            return;
        }
        state = State.CLOSED;
        if (call != null) {
            call.cancel();
        }
        listener.onClosed(this);
    }

    private void close() {
        if (state == State.CLOSED || state == State.SHUTDOWN) {
            return;
        }
        state = State.CLOSED;
        listener.onClosed(this);
    }

    @Override
    public State state() {
        return state;
    }
} 